package com.fanxuankai.boot.mqbroker.producer.autoconfigure;

import com.alibaba.fastjson.JSON;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.producer.AbstractMqProducer;
import com.fanxuankai.boot.mqbroker.producer.MqProducer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({KafkaTemplate.class})
@ConditionalOnProperty(name = MqBrokerProducerProperties.ENABLED, matchIfMissing = true)
public class MqBrokerProducerKafkaAutoConfiguration {
    @Bean
    @ConditionalOnMissingBean(MqProducer.class)
    public AbstractMqProducer mqProducer(KafkaTemplate<String, String> kafkaTemplate) {
        return new AbstractMqProducer() {
            @Override
            public boolean isPublisherCallback() {
                return false;
            }

            @Override
            public void accept(Event<String> event) {
                if (StringUtils.isEmpty(event.getDataKey())) {
                    kafkaTemplate.send(event.getName(), JSON.toJSONString(event));
                } else {
                    kafkaTemplate.send(event.getName(), event.getDataKey(), JSON.toJSONString(event));
                }
            }
        };
    }
}
